package com.gzyj.flink.gps.map.roadlevel;

import cn.hutool.core.collection.CollectionUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.alibaba.fastjson.support.spring.FastJsonRedisSerializer;
import com.gzyj.flink.driver.Driverlogininfo;
import com.gzyj.flink.dto.alarm.Kafka_VehicleRoadLevelAlarm;
import com.gzyj.flink.dto.alarm.MonitorClientAlarm;
import com.gzyj.flink.dto.utils.DateUtils;
import com.gzyj.flink.gps.config.RedisKeyPrefixConfig;
import com.gzyj.flink.gps.factory.RedisSessionFactory;
import com.gzyj.flink.pojo.gps.BaobiaoAlarmsummary;
import com.gzyj.flink.pojo.gps.BaobiaoAlarmsummaryRecord;
import com.gzyj.flink.veh.VehicleIncrement;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.StringRedisSerializer;

import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.Set;

public class KafkaRoadLevelAlarmLevelProcess extends RichMapFunction<Kafka_VehicleRoadLevelAlarm, KafkaProcessResult> {
    LettuceConnectionFactory redisFactory;


    private RedisTemplate<String, Object> getRedisTemplate(LettuceConnectionFactory redisSessionFactory) {
        RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();
        redisTemplate.setConnectionFactory(redisSessionFactory);
        redisTemplate.setKeySerializer(new StringRedisSerializer());
        redisTemplate.setValueSerializer(new FastJsonRedisSerializer<>(Object.class));
        redisTemplate.setHashKeySerializer(new StringRedisSerializer());
        redisTemplate.setHashValueSerializer(new FastJsonRedisSerializer<>(Object.class));
        redisTemplate.afterPropertiesSet();
        return redisTemplate;
    }

    public VehicleIncrement getVehicleInfo(String vehId, RedisTemplate<String, Object> clusterRedisTemplate) {

        Set<String> keys = clusterRedisTemplate.keys(RedisKeyPrefixConfig.RedisKey_VehicleInfo + "*:" + vehId + "*");
        VehicleIncrement vehicleInfo = null;
        if (CollectionUtil.isNotEmpty(keys)) {
            Object obj = clusterRedisTemplate.opsForValue().get(new ArrayList<String>(keys).get(0));
            if (obj == null) {
                return null;
            } else {
                vehicleInfo = JSONObject.parseObject(JSON.toJSONString(obj, SerializerFeature.DisableCircularReferenceDetect), VehicleIncrement.class);
            }
        } else {
            return null;
        }
        return vehicleInfo;
    }

    public Driverlogininfo TryGetData_DLD(String vehId, RedisTemplate<String, Object> clusterRedisTemplate) {
        Object obj = clusterRedisTemplate.opsForValue().get(RedisKeyPrefixConfig.RedisKey_DriverSignOn + vehId);
        if (obj == null) {
            return new Driverlogininfo();
        } else {
            return JSON.parseObject(JSON.toJSONString(obj, SerializerFeature.DisableCircularReferenceDetect), Driverlogininfo.class);
        }

    }

    @Override
    public void open(Configuration parameters) throws Exception {
        redisFactory = RedisSessionFactory.getRedisFactory();
    }

    @Override
    public void close() throws Exception {
        super.close();
    }

    @Override
    public KafkaProcessResult map(Kafka_VehicleRoadLevelAlarm RoadLevelAlarmLeve) throws Exception {
        RedisTemplate<String, Object> clusterRedisTemplate = getRedisTemplate(redisFactory);
        VehicleIncrement vehicleinfo = getVehicleInfo(RoadLevelAlarmLeve.getVehId(), clusterRedisTemplate);
        Driverlogininfo driverlogininfo=TryGetData_DLD(RoadLevelAlarmLeve.getVehId(),clusterRedisTemplate);
        KafkaProcessResult tuple3 = new KafkaProcessResult();
        if (vehicleinfo == null) return null;

        if (RoadLevelAlarmLeve.getAlarmLevel() != 99) {
            BaobiaoAlarmsummary baobiaoAlarmsummary = new BaobiaoAlarmsummary();
            baobiaoAlarmsummary.setStatus(RoadLevelAlarmLeve.getAlarmLevel());

            baobiaoAlarmsummary.setVehid(RoadLevelAlarmLeve.getVehId());
            baobiaoAlarmsummary.setDeptId(RoadLevelAlarmLeve.getDeptId());
            baobiaoAlarmsummary.setAlarmtype("超速报警");
            baobiaoAlarmsummary.setAnalyzemode(1);
            baobiaoAlarmsummary.setBegintime(RoadLevelAlarmLeve.getAlarmBeginTime());
            baobiaoAlarmsummary.setDriverName(driverlogininfo.getDriverName());
            baobiaoAlarmsummary.setDriverNo(driverlogininfo.getDriverNo());

            tuple3.setBaobiaoAlarmsummary(baobiaoAlarmsummary);
        }

        if (RoadLevelAlarmLeve.getAlarmLevel() != 99) {
            //推送到kafka
            MonitorClientAlarm vi = new MonitorClientAlarm();
            vi.VehNo = vehicleinfo.getCheliangpaizhao();
            vi.VehId = RoadLevelAlarmLeve.getVehId();
            vi.PlateColor = vehicleinfo.getChepaiyanse();
            vi.AlarmFlag = 1;
            vi.Time = DateUtils.toStr(RoadLevelAlarmLeve.getAlarmBeginTime());
            vi.Note = "超速报警";
            vi.Longitude = RoadLevelAlarmLeve.getLongitude();
            vi.Latitude = RoadLevelAlarmLeve.getLatitude();
            vi.Velocity = RoadLevelAlarmLeve.getVelocity();
            vi.Angle = RoadLevelAlarmLeve.getAngle();
            vi.Local = RoadLevelAlarmLeve.getLocal();
            vi.Limited = RoadLevelAlarmLeve.getRoadLimited();
            vi.RoadLimited = RoadLevelAlarmLeve.getRoadLimited();
            vi.RoadName = RoadLevelAlarmLeve.getRoadName();
            vi.AlarmGuid = RoadLevelAlarmLeve.getAlarmGuid();
            vi.Driver = RoadLevelAlarmLeve.getDriver();
            vi.AlarmLevel = RoadLevelAlarmLeve.getAlarmLevel();
            vi.alarmSource = 1;
            tuple3.setMonitorClientAlarm(vi);
        }
        BaobiaoAlarmsummaryRecord baobiaoAlarmsummaryRecord = new BaobiaoAlarmsummaryRecord();
        baobiaoAlarmsummaryRecord.setVehid(RoadLevelAlarmLeve.getVehId());
        baobiaoAlarmsummaryRecord.setBegintime(DateUtils.toDate(RoadLevelAlarmLeve.getTime()));
        baobiaoAlarmsummaryRecord.setAlarmtype("超速报警");
        baobiaoAlarmsummaryRecord.setLongitude(BigDecimal.valueOf(RoadLevelAlarmLeve.getLastlongitude()));
        baobiaoAlarmsummaryRecord.setLatitude(BigDecimal.valueOf(RoadLevelAlarmLeve.getLastlatitude()));
        baobiaoAlarmsummaryRecord.setVelocity(Long.valueOf(Math.round(RoadLevelAlarmLeve.getVelocity())).intValue());
        baobiaoAlarmsummaryRecord.setAngle(RoadLevelAlarmLeve.getAngle());
        baobiaoAlarmsummaryRecord.setRoadLevel(RoadLevelAlarmLeve.getRoadLevel());
        baobiaoAlarmsummaryRecord.setRoadLimited(RoadLevelAlarmLeve.getRoadLimited());
        baobiaoAlarmsummaryRecord.setKeeptime(Long.valueOf(RoadLevelAlarmLeve.getAlarmTimeSpan()).intValue());
        baobiaoAlarmsummaryRecord.setLevel(RoadLevelAlarmLeve.getAlarmLevel());
        baobiaoAlarmsummaryRecord.setAlarmreportid(RoadLevelAlarmLeve.getAlarmGuid());

        tuple3.setBaobiaoAlarmsummaryRecord(baobiaoAlarmsummaryRecord);

        return tuple3;
    }
}
